package com.crawler.actor

import java.util.concurrent.Executors

import akka.actor._
import akka.util.Timeout
import com.crawler.message.TaskParamListMessage
import com.crawler.service.DBService

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._

/**
  * Created by tongtao.zhu on 2016/12/19.
  */
class DispatcherActor extends Actor with ActorLogging {
  implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))
  implicit val timeout = Timeout(20 seconds)
  val dbService = DBService.dbService
  var cancellable: Cancellable = null

  override def postStop(): Unit = {
    log.info("结束一个DispatcherActor")
  }

  override def preStart(): Unit = {
    // 订阅集群事件
    log.info("开始一个DispatcherActor")
  }

  override def receive = {
    case source: String =>{
      val tasks = dbService.findTasks(source)
      val runningTasks = dbService.findRunningTask(source)
      val routerRef = context.actorSelection("/user/routerRef")
      if (tasks.length == 0 && runningTasks==0){
        log.info("所有的任务已经结束")
        cancellable cancel()
      }
      if (tasks.length > 0){
        routerRef ! TaskParamListMessage(tasks)
      }
      if (cancellable == null){
        schedule(source)
      }
    }
  }

  def schedule(dbName: String): Unit ={
    cancellable = context.system.scheduler.schedule(10 seconds, 10 seconds, self, dbName)
  }
}
